package com.catmiao.spark.stream

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 * @title: SparkStreaming12_MockData
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/3/26 22:14
 */
object SparkStreaming12_MockData {


  def main(args: Array[String]): Unit = {

    // 生成模拟数据
    // 时间戳 区域 城市 用户 广告
    // timestamp area city userid adid

    // Application => Kafka => SparkStreaming => Analysis

    // 创建配置对象
    val prop = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")

    val kafkaProducer: KafkaProducer[String, String] = new KafkaProducer[String, String](prop)
    while (true) {
      mockData().foreach(
        data => {
          val record: ProducerRecord[String, String] = new ProducerRecord[String, String]("first", data)
          // 向kafka中生成数据
          kafkaProducer.send(record)
          //          println(record)
        }
      )
      Thread.sleep(2000)
    }
  }

  def mockData(): ListBuffer[String] = {
    val list = ListBuffer[String]()
    val areaList = ListBuffer[String]("华东", "华北", "华南")
    val cityList = ListBuffer[String]("北京", "上海", "广州")
    for (i <- 1 to 30) {
      val area = areaList(new Random().nextInt(3))
      val city = cityList(new Random().nextInt(3))
      val userId = new Random().nextInt(6) + 1
      val adid = new Random().nextInt(6) + 1

      list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userId} ${adid}")
    }

    list
  }
}
